package com.example.demo.config.artemis;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.*;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
import org.apache.activemq.artemis.core.server.embedded.EmbeddedActiveMQ;
import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;

import java.util.HashMap;
import java.util.Map;

public final class ActiveMqManager {
    private static final Logger logger = LoggerFactory.getLogger(ActiveMqManager.class);
    public static final String PROTO_VM = "vm://0";
    public static final String PROTO_TCP = "tcp://127.0.0.1:61616";
    public static final String RDO_QUEUE_VM = "RDO_QUEUE_VM";
    public static final String RDO_QUEUE_TCP = "RDO_QUEUE_TCP";
    public static final String RDOP_QUEUE = "RDOP_QUEUE";
    public static final String RDOE_QUEUE = "RDOE_QUEUE";
    public static final String RDOI_QUEUE = "RDOI_QUEUE";

    private static EmbeddedActiveMQ server = new EmbeddedActiveMQ();

    private static ClientSessionFactory factoryVm = null;
    private static ClientSessionFactory factoryTcp = null;


    private ActiveMqManager() {

    }

    public static ClientProducer createNewProducer(ClientSession session, String queueName) {
        try {
            return session.createProducer(queueName);
        } catch (ActiveMQException e) {
            logger.error("Failed to create Producer, queue name : {}, reason: {}", queueName, e.getMessage());
        }

        return null;
    }

    public static ClientConsumer createNewConsumer(ClientSession session, String queueName) {
        try {
            return session.createConsumer(queueName);
        } catch (ActiveMQException e) {
            logger.error("Failed to create Producer, queue name : {}, reason: {}", queueName, e.getMessage());
        }

        return null;
    }

    public static void createQueue(ClientSession session, String queueName) {
        try {
            ClientSession.QueueQuery queueQuery = session.queueQuery(new SimpleString(queueName));
            if(!queueQuery.isExists()) {
                session.createQueue(new QueueConfiguration(queueName));
            }
        } catch (ActiveMQException e) {
            logger.error("Failed to create queue, queue name : {}, reason: {}", queueName, e.getMessage());
        }
    }

    public static void sendMessage(ClientSession session, ClientProducer producer, String msg, Boolean durable) {
        try {
            ClientMessage message = session.createMessage(durable);
            message.getBodyBuffer().writeString(msg);
            producer.send(message);
        } catch (Exception e) {
            logger.error("Failed to send message : {}, reason: {}", msg, e.getMessage());
        }
    }


    public static String getMessageVm(ClientConsumer consumer, Long timeout, Boolean immediate) {
        try {
            ClientMessage msgReceived = null;
            if(null != timeout && timeout > 3000 && timeout < 300 * 1000) {
                msgReceived = consumer.receive(timeout);
            } else if(immediate == Boolean.TRUE) {
                msgReceived = consumer.receiveImmediate();
            } else {
                msgReceived = consumer.receive();
            }
            return msgReceived.getBodyBuffer().readString();
        } catch (Exception e) {
            logger.error("Failed to get message, reason: {}", e.getMessage());
        }
        return null;
    }
    public static String getMessage(ClientConsumer consumer) {
        return getMessageVm(consumer, null, null);
    }

    public static void setMessageHandler(ClientConsumer consumer, MessageHandler handler) {
        try {
            consumer.setMessageHandler(handler);
        } catch (ActiveMQException e) {
            logger.error("Failed to setMessageHandler, reason: {}", e.getMessage());
        }
    }

    public static void startServer() {
        logger.info("ActiveMQ START");
        org.apache.activemq.artemis.core.config.Configuration config = new ConfigurationImpl();
        LoggingActiveMQServerPlugin logPlugin = new LoggingActiveMQServerPlugin();
        Map<String, String> props = new HashMap<>();
        props.put("LOG_ALL_EVENTS", "true");
        logPlugin.init(props);
        config.registerBrokerPlugin(logPlugin);

        try {
            config.addAcceptorConfiguration("in-vm", PROTO_VM);
            config.addAcceptorConfiguration("tcp", PROTO_TCP);
            config.setSecurityEnabled(false);
            config.setPersistenceEnabled(false);
            server.setConfiguration(config);
            server.start();

        } catch (Exception e) {
            logger.error("Failed to start ActiveMQ, reason: {}", e.getMessage());
        }

        initClientSessionFactory();
    }

    public static void initClientSessionFactory() {
        factoryVm = createClientSessionFactory(PROTO_VM);
        factoryTcp = createClientSessionFactory(PROTO_TCP);
    }

    private static ClientSessionFactory createClientSessionFactory(String proto) {
        ServerLocator serverLocator = null;
        try {
            serverLocator = ActiveMQClient.createServerLocator(proto);
            ClientSessionFactory factory = serverLocator.createSessionFactory();


            return factory;
        } catch (Exception e) {
            logger.error("Failed to create client session factory, reason: {}", e.getMessage());
        }

        return null;
    }

    public static ClientSession createClientSession(String proto, Boolean start) {
        try {
            ClientSession session = null;
            if(PROTO_VM.equals(proto)) {
                session = factoryVm.createSession();
            } else if(PROTO_TCP.equals(proto)) {
                //session = factoryTcp.createSession();
                session = factoryTcp.createSession("admin", "rocket", false,
                        true,
                        true,
                        ActiveMQClient.DEFAULT_PRE_ACKNOWLEDGE,
                        ActiveMQClient.DEFAULT_ACK_BATCH_SIZE, "MY-SESSION-ID");
            } else {
                session = factoryVm.createSession();
            }

            if(start == Boolean.TRUE) {
                session.start();
            }

            return session;
        } catch (Exception e) {
            logger.error("Failed to get client MQ session, reason: {}", e.getMessage());
        }

        return null;
    }

    private static void closeSessionFactory(ClientSessionFactory factory) {
        if(factory != null && !factory.isClosed()) {
            try {
                factory.close();
            } catch (Exception e) {
                logger.error("Failed to close client session factory, reason: {}", e.getMessage());
            }
        }
    }

    public static void closeSession(ClientSession session) {
        if(session != null && !session.isClosed()) {
            try {
                session.close();
            } catch (Exception e) {
                logger.error("Failed to close client session, reason: {}", e.getMessage());
            }
        }
    }

    public static void stopServer() {
        logger.info("ActiveMQ STOP");

        closeSessionFactory(factoryVm);
        closeSessionFactory(factoryTcp);

        try {
            server.stop();
        } catch (Exception e) {
            logger.error("Failed to stop ActiveMQ, reason: {}", e.getMessage());
        }
    }
}
